package com.bw.flinkstreaming.state1.job3;

import com.bw.flinkstreaming.state.job2.ListStateWithCountAvg;
import com.bw.flinkstreaming.state1.job2.ValueStateWithCountAvg;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * （hadoop,1）（hadoop,11）（hadoop,20） 20 +11 + 1 / 3
 *
 *   需求：当接收到的相同 key 的元素个数等于 3 个或者超过 3 个的时候
 *  *  就计算这些元素的 value 的平均值。
 *    计算 keyed stream 中每 3 个元素的 value 的平均值
 */

public class ListStateWithCountAvgTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //组织数据源
        DataStreamSource<Tuple2<Long, Long>> dataStreamSource = env.fromElements(Tuple2.of(10L, 64L),Tuple2.of(12L, 16L), Tuple2.of(12L, 85L), Tuple2.of(12L, 11L), Tuple2.of(10L, 33L), Tuple2.of(10L, 88L), Tuple2.of(13L, 88L));
        SingleOutputStreamOperator<Tuple2<Long, Double>> tuple2SingleOutputStreamOperator = dataStreamSource.keyBy(0).flatMap(new ListStateWithCountAvg());
        tuple2SingleOutputStreamOperator.print();
        env.execute("ListStateWithCountAvgTest");
    }
}
